package day9;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Properties;

import static org.apache.flink.table.api.Expressions.$;

public class Test5_kafka_to_mysql {
    public static void main(String[] args) throws Exception {
        // 从kafka读数据(csv文件)到mysql
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        TableResult table1 = tenv.executeSql(
                "CREATE TABLE table1 (\n" +
                        "  `name` STRING,\n" +
                        "  `age` INT,\n" +
                        "  `status`  STRING\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'topic1',\n" +
                        "  'properties.bootstrap.servers' = 'hadoop10:9092',\n" +
                        "  'properties.group.id' = 'testGroup',\n" +
                        "  'scan.startup.mode' = 'latest-offset',\n" +
                        "  'format' = 'csv'\n" +
                        ")"
        );

        //向MYSQL写数据
        tenv.executeSql(
                "CREATE TABLE table2 (\n" +
                        "    name               VARCHAR,  \t\n" +
                        "  `age`		  BIGINT,\n" +
                        "  `status`  		  STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'jdbc',\n" +
                        "    'url' = 'jdbc:mysql://hadoop10:3306/test1?useUnicode=true&characterEncoding=utf8',\n" +
                        "    'table-name' = 't_success', \n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );

        env.execute();
    }
}
